package com.lxn.app

import com.alibaba.fastjson.serializer.SerializeConfig
import com.alibaba.fastjson.{JSON, JSONObject}
import com.lxn.bean.{OrderDetail, OrderInfo, OrderWide}
import com.lxn.bean
import com.lxn.until.{MYKafkaUtils, MyEsUtils, MyOffsetUtils, MyRedisUtils}
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.kafka.common.TopicPartition
import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.{DStream, InputDStream}
import org.apache.spark.streaming.kafka010.{HasOffsetRanges, OffsetRange}
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.codehaus.jackson.map.SerializationConfig
import redis.clients.jedis.Jedis

import java.util
import java.time.{LocalDate, Period}
import java.util.Date
import scala.Option
import scala.collection.mutable.ListBuffer
import scala.language.postfixOps

object DWDOrderApp {

  //其实过程都差不多
  //这个项目是处理两个双流join合并成一个订单表
  def main(args: Array[String]): Unit = {
    val sparkConf = new SparkConf().setAppName("dwd_order_app").setMaster("local[4]")
    val ssc = new StreamingContext(sparkConf, Seconds(5))

    //读取偏移量
    val orderInfoTopicName : String = "DWD_ORDER_INFO_I"
    val orderInfoGroup : String = "DWD_ORDER_INFO_GROUP"

    val orderInfoOffsets: Map[TopicPartition, Long] = MyOffsetUtils.readOffset(orderInfoGroup, orderInfoTopicName)

    //读取detail表
    val orderDetailTopicName : String = "DWD_ORDER_DETAIL_I"
    val orderDetailGroup : String = "DWD_ORDER_DETAIL_GROUP"

    val orderDetailOffsets : Map[TopicPartition,Long] = MyOffsetUtils.readOffset(orderDetailGroup,orderDetailTopicName)

    //获取消费
    var orderInfoOffsetsDStream: InputDStream[ConsumerRecord[String, String]] = null
    if(orderInfoOffsets!=null && orderDetailOffsets.nonEmpty){
      orderInfoOffsetsDStream = MYKafkaUtils.getConsumer(ssc, orderInfoTopicName, orderInfoGroup,orderInfoOffsets)
    }else{
      orderInfoOffsetsDStream = MYKafkaUtils.getConsumer(ssc, orderInfoTopicName, orderInfoGroup)
    }

    var orderDetailOffsetsDStream: InputDStream[ConsumerRecord[String, String]] = null
    if(orderDetailOffsets!=null && orderDetailOffsets.nonEmpty){
      orderDetailOffsetsDStream= MYKafkaUtils.getConsumer(ssc, orderDetailTopicName, orderDetailGroup, orderDetailOffsets)
    }else{
      orderDetailOffsetsDStream= MYKafkaUtils.getConsumer(ssc, orderDetailTopicName, orderDetailGroup)
    }

    //提取结束点偏移量
    var orderInfoOffsetRanges: Array[OffsetRange] = null
    val orderInfoOffsetRangesDStream : DStream[ConsumerRecord[String,String]] = orderInfoOffsetsDStream.transform(
      rdd => {
        orderInfoOffsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges
        rdd
      }
    )

    var orderDetailOffsetRanges: Array[OffsetRange] = null
    val orderDetailOffsetRangesDStream : DStream[ConsumerRecord[String,String]] = orderDetailOffsetsDStream.transform(
      rdd => {
        orderDetailOffsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges
        rdd
      }
    )

    //处理数据
    //转换为专用类型
    val orderInfoDStream = orderInfoOffsetRangesDStream.map(
      consumerRecord => {
        val value: String = consumerRecord.value()
        //转换数据格式
        val orderInfoValue = JSON.parseObject(value, classOf[OrderInfo])
        orderInfoValue
      }
    )
//    orderInfoDStream.print(100)

    val orderDetailDStream = orderDetailOffsetRangesDStream.map(
      consumerRecord => {
        val value: String = consumerRecord.value()
        val orderDetailValue = JSON.parseObject(value, classOf[OrderDetail])
        orderDetailValue
      }
    )
//    orderDetailDStream.print(100)

    //关联数据
    val orderInfoMapParDStream : DStream[OrderInfo] = orderInfoDStream.mapPartitions(
      orderInfoIter => {
        val orderInfoList: List[OrderInfo] = orderInfoIter.toList
        //        val orderInfos: ListBuffer[OrderInfo] = ListBuffer[OrderInfo]()
        val jedis: Jedis = MyRedisUtils.getRedisFromPool()
        for (orderInfo <- orderInfoList) {

          //提取数据
          val uid: Long = orderInfo.user_id
          //拼接key，获取redis中的数据
          val redisKey: String = s"DIM:USER_INFO:$uid"
          val userInfoJson: String = jedis.get(redisKey)
          //字符串不能提取数据，转换为Json
          val userInfoJsonObj: JSONObject = JSON.parseObject(userInfoJson)
          val gender: String = userInfoJsonObj.getString("gender")
          val birthday: String = userInfoJsonObj.getString("birthday")
          //将是生日转换为年龄
          val birthdayLd: LocalDate = LocalDate.parse(birthday)
          val dateNow: LocalDate = LocalDate.now()
          val period = Period.between(birthdayLd, dateNow)
          val age: Int = period.getYears

          //封装字段
          orderInfo.user_gender = gender
          orderInfo.user_age = age


          //地区维度
          val provinceId: Long = orderInfo.province_id
          val provinceRedisKey: String = s"DIM:BASE_PROVINCE:$provinceId"
          val provinceJson: String = jedis.get(provinceRedisKey)
          val provinceJsonObj: JSONObject = JSON.parseObject(provinceJson)
          val provinceName: String = provinceJsonObj.getString("name")
          val provinceAreaCode: String = provinceJsonObj.getString("area_code")
          val province3166: String = provinceJsonObj.getString("ios_3166_2")
          val provinceIosCode: String = provinceJsonObj.getString("iso_code")

          orderInfo.province_name = provinceName
          orderInfo.province_iso_code = provinceIosCode
          orderInfo.province_3166_2_code = province3166
          orderInfo.province_area_code = provinceAreaCode

          //获取时间，天，小时
          //观察数据，我们发现数据里面有时间，还是格式化时间字符串，所以可以之间切分字符串
          val timeStr = orderInfo.create_time.split(" ")
          val day: String = timeStr(0)
          val hour: String = timeStr(1).split(":")(0)

          orderInfo.create_date = day
          orderInfo.create_hour = hour

          //保存一条数据
          //          orderInfos.append(orderInfo)


        }
        jedis.close()
        orderInfoList.iterator
      }
    )
//    orderInfoMapParDStream.print(100)

    //好了，现在我们有两张表的完整数据了，现在考虑把两张表的数据进行双流join
    //由于订单详情表是来自订单表，所以就算是使用内连接也是能够将两个表的数据全部join
    //考虑使用什么join 可以内join 也可以外join 外join又分为left right full join
    //由于join是在流中完成的，所以如果数据存在延迟，就会导致数据丢失，所以我们选择全外连接，但是延迟还是会导致数据无法join ，这样我们只是保证数据不会丢失
    //转换数据格式，才能进行join，需要指定关联字段
    val orderInfoKVDStream :DStream[(Long,OrderInfo)]
    = orderInfoMapParDStream.map(OrderInfo => (OrderInfo.id, OrderInfo))

    val orderDetailKVDStream :DStream[(Long,OrderDetail)]
    = orderDetailDStream.map(OrderDetail => (OrderDetail.order_id, OrderDetail))

    //进行join
    val orderJoinDStream : DStream[(Long,(Option[OrderInfo],Option[OrderDetail]))]
    = orderInfoKVDStream.fullOuterJoin(orderDetailKVDStream)

    //然后分情况讨论，一共有三种情况 两个都有，或者其中一个没有
    //还是那句话，分布式一个分区有一股流，所以得按分区来写
    val orderWideDStream :DStream[OrderWide] = orderJoinDStream.mapPartitions(
      orderWideIter => {
        //创建容器用于存储
        val orderWides: ListBuffer[OrderWide] = ListBuffer[OrderWide]()
        val jedis: Jedis = MyRedisUtils.getRedisFromPool()
        for ((key, (orderInfoOp, orderDetailOp)) <- orderWideIter) {
          println(s"[JOIN_DEBUG] Key: $key, OrderInfo exists: ${orderInfoOp.isDefined}, OrderDetail exists: ${orderDetailOp.isDefined}")
          // 其他处理逻辑
          //进行值判断，是哪一种情况先处理两种都存在的情况
          //因为Option这种值不会为null,只会存在有无数据，所以判断条件和一般判断有无有点区别
          //这种数据类型就是用一个不为null的变量保存，值在它肚子里，直接get出来就好了
          if (orderInfoOp.isDefined) {
            //直接取出数据
            val orderInfo: OrderInfo = orderInfoOp.get
            if (orderDetailOp.isDefined) {
              val orderDetail: OrderDetail = orderDetailOp.get
              //然后将值保存到组合表
              val orderWide = new OrderWide(orderInfo, orderDetail)
              orderWides.append(orderWide)
            }
            //由于在if外面，所以if过滤掉了符合它的逻辑，所以外面就是它相反的逻辑
            //也就是orderDetailOp不存在

            //那我们就需要写缓存和读缓存 因为我们采用让两表能够保证数据全部join是通过等待的，在一定时间内，我会将未关联的数据存储到redis中
            //然后后面来的，都会去查看redis中的数据，判断是否存在可关联的数据
            //存储方式，订单表的存储方式是一个一个存储，详情表存储时按id一堆存储，这样就可以让一个订单关联多个详情数据

            //写数据 将未关联的orderInfo存储
            val orderInfoRedisKey: String = s"ORDERJOIN:ORDER_INFO:${orderInfo.id}"
            //存储数据，老样子，专用对象需指定使用字段
            jedis.setex(orderInfoRedisKey, 24 * 3600, JSON.toJSONString(orderInfo, new SerializeConfig(true)))


            //读数据
            //我们使用set储存详情，然后，我们需要根据相同的key寻找响应的详情信息进行join
            val orderDetailRedisKey: String = s"ORDERJOIN:ORDER_DETAIL:${orderInfo.id}"
            //根据key获取值
            val orderDetailStr: util.Set[String] = jedis.smembers(orderDetailRedisKey)
            //首先我们需要判断这个是否为空，是否存在数据
            if (orderDetailStr != null && orderDetailStr.size() > 0) {
              //将数据转化为可以join的专用类型
              //转换成专用类型
              //由于这个获取到的是java的Set集合，所以我们不能直接对它进行遍历，我们使用这个隐式转换为scala可以遍历的集合
              import scala.collection.JavaConverters._
              for (orderDetailObj <- orderDetailStr.asScala) {
                val orderDetail = JSON.parseObject(orderDetailObj, classOf[OrderDetail])
                //然后将获取的数据，与当前批次的orderInfo进行join
                val orderWide: OrderWide = new OrderWide(orderInfo, orderDetail)
                //将数据保存
                orderWides.append(orderWide)

              }
            }

          } else {
            //因为这两个表进行join它最起码会有一个存在，所以这里肯定是只有orderDetail存在
            //这里就是orderInfo不存在，然后我们只对orderDetail存在时进行处理
            //这就是orderInfo不存在 orderDetail存在的情况
            //直接获取orderDetail数据
            val orderDetail: OrderDetail = orderDetailOp.get
            //将数据转化为可以发送的字符串
            //读数据
            //这个因为它只能关联一次，所以先关联在存储，这样就减少了redis的存储
            val orderInfoRedisKey: String = s"ORDERJOIN:ORDER_INFO:${orderDetail.order_id}"
            val orderInfoStr: String = jedis.get(orderInfoRedisKey)
            //判断值是否为空，或者没有数据的情况
            if (orderInfoStr != null && orderInfoStr.size > 0) {
              //数据不为空，读数据
              //转化数据类型
              val orderInfo: OrderInfo = JSON.parseObject(orderInfoStr, classOf[OrderInfo])
              //封装数据
              val orderWide = new OrderWide(orderInfo, orderDetail)
              orderWides.append(orderWide)
            } else {

              //写数据，将数据存储至redis
              val orderDetailRedisKey: String = s"ORDERJOIN:ORDER_DETAIL:${orderDetail.order_id}"
              //存储数据
              jedis.sadd(orderDetailRedisKey, JSON.toJSONString(orderDetail,new SerializeConfig(true)))
              jedis.expire(orderDetailRedisKey, 24 * 3600)

            }
          }

        }
        jedis.close()
        orderWides.iterator
      }
    )

    orderWideDStream.print(100)

    //将数据写入es
    orderWideDStream.foreachRDD(
      rdd =>{
        //对数据进行处理
        rdd.foreachPartition(
          //对每一个迭代对象进行数据转换,因为每个分区都可能有值，所以返回的是一个迭代对象
          orderWideIter =>{
            //转换数据格式
            val docs: List[(String, OrderWide)]
            = orderWideIter.map(orderWide => (orderWide.detail_id.toString, orderWide)).toList
            //判断数据是否为空
            if(docs.size>0){
              //我们需要传入索引名，幂等性操作的id以及数据
              //拼接索引名称
              //先获取时间
              //获取第一行数据
              val head = docs.head
              val dateStr :String = head._2.create_date
              val indexName :String = s"gmall_order_wide$dateStr"
              //封装数据
              MyEsUtils.bulkSave(indexName,docs)
            }
          }
        )
        //保存偏移量
        MyOffsetUtils.saveOffset(orderInfoGroup,orderInfoTopicName,orderInfoOffsetRanges)
        MyOffsetUtils.saveOffset(orderDetailGroup,orderDetailTopicName,orderDetailOffsetRanges)
      }
    )

    ssc.start()
    ssc.awaitTermination()

  }

}
